package com.sumbo.config.writer;

import java.util.Map;
import java.util.Properties;

/**
 * \* User: MeiZhongHao
 * \* Date: 2019-06-10
 * \* Time: 17:01
 * \* Description: 输出工厂类
 * \
 */
public class WriterFactory {

    public static void init() {
        // 初始化kafkaWriter
        Map<String, Properties> kakfaWriterConfig = WriterConfig.getKafkaConfig();
        KafkaWriter.instance.init(kakfaWriterConfig);
        // 初始化rocketMqWriter
        Map<String, Properties> rocketMqConfig = WriterConfig.getRocketMqConfig();
        RocketMqWriter.instance.init(rocketMqConfig);
    }

    public static void send(OutputContext out, String msg) {
        switch (out.getType()) {
            case KAFKA:
                KafkaWriter.instance.send((KafkaOutputContext) out, msg);
                break;
            case ROCKETMQ:
                RocketMqWriter.instance.send((RocketMqOutputContext) out, msg);
                break;
        }
    }

    public static OutputContext analysis(String client, Properties properties) {
        OutputContext context = null;
        if (KafkaWriter.instance.containsProducer(client)) {
            context = analysisKafka(properties);
        } else if (RocketMqWriter.instance.containsProducer(client)) {
            context = analysisRocketMq(properties);
        }
        context.setClient(client);
        return context;
    }

    private static OutputContext analysisRocketMq(Properties properties) {
        RocketMqOutputContext context = new RocketMqOutputContext();
        context.setTag(properties.getProperty("tag"));
        context.setTopic(properties.getProperty("topic"));
        return context;
    }

    private static KafkaOutputContext analysisKafka(Properties properties) {
        KafkaOutputContext context = new KafkaOutputContext();
        context.setTopic(properties.getProperty("topic"));
        return context;
    }

}